package org.example.tools;

import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.RestOptions;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * Flink WebUI 演示类
 * 这个类专门用于展示如何启用和使用Flink的WebUI界面
 * 演示了创建带有WebUI的Flink环境，并执行简单的表操作
 * 执行完SQL后保持程序运行，以便继续使用WebUI
 */
public class FlinkWebUIDemo {
    public static void main(String[] args) throws Exception {
        // 创建配置对象，设置Web UI端口
        Configuration configuration = new Configuration();
        configuration.set(RestOptions.PORT, 8081);
        
        // 使用配置创建执行环境，启用WebUI
        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(configuration);
        // 设置并行度为1
        env.setParallelism(1);
        // 开启checkpoint，并设置间隔为5秒
        env.enableCheckpointing(5000);
        env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
        
        // 创建StreamTableEnvironment
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        // 打印环境信息
        System.out.println("Flink WebUI 已启动，请访问 http://localhost:8081");
        System.out.println("环境配置信息：");
        System.out.println("- 并行度: " + env.getParallelism());
        System.out.println("- 执行模式: STREAMING");
        System.out.println("- Checkpoint间隔: 5秒");
        
        // 创建一个使用datagen的表，用于测试和展示
        System.out.println("创建测试表: datagen_test_table");
        String createTableSQL = "CREATE TABLE datagen_test_table (\n" +
                "  id INT,\n" +
                "  name STRING,\n" +
                "  score DOUBLE,\n" +
                "  event_time TIMESTAMP(3)\n" +
                ") WITH (\n" +
                "  'connector' = 'datagen',\n" +
                "  'rows-per-second' = '10'\n" +
                ")";
        
        System.out.println("执行SQL：");
        System.out.println(createTableSQL);
        tableEnv.executeSql(createTableSQL);
        
        // 查询数据并打印
        String querySQL = "SELECT * FROM datagen_test_table";
        System.out.println("执行SQL查询：");
        System.out.println(querySQL);
        tableEnv.executeSql(querySQL).print();
    }
} 